package com.tsfyun.scm.system.config;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.tsfyun.scm.system.service.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @Description: mysql binlog监听，未考虑多台mysql断点续传问题，可以记录断点位置
 * @CreateDate: Created in 2021/9/17 15:10
 */
@Component
@Order(value = 1000)
public class MysqlBinlogWatch implements ApplicationRunner{

    @Value("${mysql.ip}")
    private String mysqlIp;
    @Value("${mysql.port}")
    private int mysqlPort;
    @Value("${mysql.username}")
    private String mysqlUser;
    @Value("${mysql.password}")
    private String mysqlPassword;

    @Autowired
    private ICountryService countryService;
    @Autowired
    private ICustomsDataService customsDataService;
    @Autowired
    private ICustomsElementsService customsElementsService;
    @Autowired
    private ICustomsCodeService customsCodeService;
    @Autowired
    private IBaseDataService baseDataService;

    //MySQL表id和表名映射
    public static Map<Long,String> MYSQL_TABLE_ID_NAME_MAP = Maps.newHashMap();
    //需要处理的表名        国家、海关编码、海关基础数据、申报要素、基础数据
    public static List<String> HAND_TABLE_NAME = Lists.newArrayList("country","customs_code"
            ,"customs_data","customs_elements","base_data");
    /**
     * 监听
     */
    public void watch() {
        BinaryLogClient client = new BinaryLogClient(mysqlIp,mysqlPort,"scm_base",mysqlUser,mysqlPassword);
        client.setEventDeserializer(new EventDeserializer());
        client.registerEventListener(event -> {
            EventHeader header = event.getHeader();
            EventType eventType = header.getEventType();
            if(Objects.equals(eventType,EventType.TABLE_MAP)) {//表id和表名映射
                TableMapEventData tableMapEventData = event.getData();
                MYSQL_TABLE_ID_NAME_MAP.computeIfAbsent(tableMapEventData.getTableId(),key->tableMapEventData.getTable());
            }
            Long tableId = null;
            if (EventType.isWrite(eventType)) {
                WriteRowsEventData data = event.getData();
                tableId = data.getTableId();
            } else if (EventType.isUpdate(eventType)) {
                UpdateRowsEventData data = event.getData();
                tableId = data.getTableId();
            } else if (EventType.isDelete(eventType)) {
                DeleteRowsEventData data = event.getData();
                tableId = data.getTableId();
            }
            if(Objects.nonNull(tableId) && MYSQL_TABLE_ID_NAME_MAP.containsKey(tableId)
                    && HAND_TABLE_NAME.contains(MYSQL_TABLE_ID_NAME_MAP.get(tableId))) {
                String tableName = MYSQL_TABLE_ID_NAME_MAP.get(tableId);
                switch (tableName) {
                    case "country" :
                        countryService.clear();
                        break;
                    case "customs_data":
                        customsDataService.clear();
                        break;
                    case "customs_elements":
                        customsElementsService.clear();
                        break;
                    case "customs_code":
                        customsCodeService.clear();
                        break;
                    case "base_data":
                        baseDataService.clear();
                        break;
                    default:
                        break;
                }
            }
        });
        try {
            client.connect();
        } catch (Exception e) {
        }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        watch();
    }
}
